package com.atguigu.canal;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.atguigu.canal.utils.MyKafkaSender;
import com.atguigu.common.logger.GmallConstants;

import java.util.List;

/**
 * 对获取的Canal数据进行封装
 *
 * @author WangJX
 * @version 1.0
 * @date 2019/11/28 10:20
 */
public class CanalHandler {
    private String tableName;
    private CanalEntry.EventType eventType;
    private List<CanalEntry.RowData> rowDatasList;

    public CanalHandler(String tableName, CanalEntry.EventType eventType, List<CanalEntry.RowData> rowDatasList) {
        this.tableName = tableName;
        this.eventType = eventType;
        this.rowDatasList = rowDatasList;
    }


    //封装方法只去读order_info表中insert的数据，并发送到kafka
    public void handle() {
        //新增订单数据
        if (tableName.equalsIgnoreCase("order_info")
                && eventType.equals(CanalEntry.EventType.INSERT)) {

            //将数据发送到GMALL_NEW_ORDER主题
            rowDateToKafka(GmallConstants.KAFKA_TOPIC_NEW_ORDER);

        } else if (tableName.equalsIgnoreCase("order_detail")
                && eventType.equals(CanalEntry.EventType.INSERT)) {     //获取订单表的新增数据

            //将数据发送到gmall_sale_detail主题
            rowDateToKafka(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL);

        } else if (tableName.equalsIgnoreCase("user_info")
                && (eventType.equals(CanalEntry.EventType.INSERT)
                || eventType.equals(CanalEntry.EventType.UPDATE))) {    //获取每日用户表的变化和新增数据
            //将数据发送到gmall_new_user主题
            rowDateToKafka(GmallConstants.KAFKA_TOPIC_NEW_USER);
        }
    }


    //发送数据到kafka
    private void rowDateToKafka(String topic) {
        for (CanalEntry.RowData rowData : rowDatasList) {
            //获取新插入的数据
            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
            JSONObject jsonObject = new JSONObject();
            //对数据进行解析
            for (CanalEntry.Column column : afterColumnsList) {
                jsonObject.put(column.getName(), column.getValue());

                System.out.println(column.getName() + "->" + column.getValue());
            }

            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            //往kafka中发送数据
            MyKafkaSender.send(topic, jsonObject.toJSONString());
            //System.out.println(jsonObject.toJSONString());
        }

    }


}
